package com.example.websocket.redis.queue;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RFuture;
import org.redisson.api.RedissonClient;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;

import java.util.Map;

/**
 * 初始化队列监听
 *
 */
//@Component
public class RedisDelayedQueueInit implements ApplicationContextAware {
	
	private static final Logger LOGGER = LogManager.getLogger(RedisDelayedQueueInit.class);

	@Autowired
	private RedissonClient redissonClient;
	
	@Override
	@SuppressWarnings({ "unchecked", "rawtypes" })
	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
		Map<String, RedisDelayedQueueListener> map = applicationContext.getBeansOfType(RedisDelayedQueueListener.class);
		for (Map.Entry<String, RedisDelayedQueueListener> taskEventListenerEntry : map.entrySet()) {
			 String listenerName = taskEventListenerEntry.getValue().getClass().getName();
			 startThread(listenerName, taskEventListenerEntry.getValue());
		}
	}
	
	
	/**
	 * 获取应用上下文并获取相应的接口实现类
	 * @param queueName
	 * @param redisDelayedQueueListener
	 */
	private <T> void startThread(String queueName, RedisDelayedQueueListener<T> redisDelayedQueueListener) {
		RBlockingQueue<T> blockingFairQueue = redissonClient.getBlockingQueue(queueName);
		// 服务重启后，无offer,take不到信息
		redissonClient.getDelayedQueue(blockingFairQueue);
		//由于此线程需要常驻，可以新建线程，不用交给线程此管理
		System.out.println("这里-----"+queueName);
		Thread thread = new Thread(() -> {
			LOGGER.info("start listener queue thread: {}", queueName);
			while (true) {
				try {
					RFuture<T> t = blockingFairQueue.takeAsync().sync();
					t.whenComplete((result, error) -> {
						if (error == null) {
							redisDelayedQueueListener.invoke(result);
						}
					});
				} catch (Exception e) {
					LOGGER.error("listener queue error:{}", e.getMessage(), e);
				}
			}
		});
		thread.setName(queueName);
		thread.start();
	}

}
